﻿using Misakai.Kafka;
using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading;
using System.Threading.Tasks;

namespace Reflexive.IO.Kafka
{
    public class KafkaConnection : IKafkaConnection
    {
        private class AsyncRequestItem
        {
            public int CorrelationId
            {
                get;
                private set;
            }
            public TaskCompletionSource<byte[]> ReceiveTask
            {
                get;
                private set;
            }
            public DateTime CreatedOnUtc
            {
                get;
                set;
            }
            public AsyncRequestItem(int correlationId)
            {
                this.CorrelationId = correlationId;
                this.CreatedOnUtc = DateTime.UtcNow;
                this.ReceiveTask = new TaskCompletionSource<byte[]>();
            }
        }
        private const double DefaultResponseTimeoutMs = 30000.0d;
        private readonly ConcurrentDictionary<int, KafkaConnection.AsyncRequestItem> _requestIndex = new ConcurrentDictionary<int, KafkaConnection.AsyncRequestItem>();
        private readonly IScheduledTimer _responseTimeoutTimer;
        private readonly TimeSpan _responseTimeoutMS;
        private readonly IKafkaLog _log;
        private readonly IKafkaTcpSocket _client;
        private readonly CancellationTokenSource _disposeToken = new CancellationTokenSource();
        private readonly SemaphoreSlim _timeoutSemaphore = new SemaphoreSlim(1, 1);
        private int _disposeCount;
        private Task _connectionReadPollingTask;
        private int _ensureOneActiveReader;
        private int _correlationIdSeed;
        public bool ReadPolling
        {
            get
            {
                return this._ensureOneActiveReader >= 1;
            }
        }
        public KafkaEndpoint Endpoint
        {
            get
            {
                return this._client.Endpoint;
            }
        }
        public KafkaConnection(IKafkaTcpSocket client, TimeSpan? responseTimeoutMs = null, IKafkaLog log = null)
        {
            this._client = client;
            this._log = (log ?? new DefaultTraceLog());
            this._responseTimeoutMS = (responseTimeoutMs ?? TimeSpan.FromMilliseconds(DefaultResponseTimeoutMs));
            this._responseTimeoutTimer = new ScheduledTimer().Do(new Action(this.ResponseTimeoutCheck)).Every(TimeSpan.FromMilliseconds(100.0)).StartingAt(DateTime.Now.Add(this._responseTimeoutMS)).Begin();
            this.StartReadStreamPoller();
        }
        public async Task<List<T>> SendAsync<T>(IKafkaRequest<T> request)
        {
            request.Correlation = this.NextCorrelationId();
            KafkaConnection.AsyncRequestItem asyncRequestItem = new KafkaConnection.AsyncRequestItem(request.Correlation);
            if (!this._requestIndex.TryAdd(request.Correlation, asyncRequestItem))
            {
                throw new ApplicationException("Failed to register request for async response.");
            }
            BinaryStream binaryStream = new BinaryStream();
            request.Encode(binaryStream);
            await this._client.WriteAsync(binaryStream);
            byte[] payload = await asyncRequestItem.ReceiveTask.Task;
            return request.Decode(payload).ToList<T>();
        }
        public override bool Equals(object obj)
        {
            return !object.ReferenceEquals(null, obj) && (object.ReferenceEquals(this, obj) || (!(obj.GetType() != base.GetType()) && this.Equals((KafkaConnection)obj)));
        }
        protected bool Equals(KafkaConnection other)
        {
            return object.Equals(this._client.Endpoint, other.Endpoint);
        }
        public override int GetHashCode()
        {
            if (this._client.Endpoint == null)
            {
                return 0;
            }
            return this._client.Endpoint.GetHashCode();
        }
        private void StartReadStreamPoller()
		{
			this._connectionReadPollingTask = Task.Factory.StartNew<Task>(async ()=>
			{
				try
				{
					if (Interlocked.Increment(ref this._ensureOneActiveReader) == 1)
					{
						while (!this._disposeToken.IsCancellationRequested)
						{
							try
							{
								byte[] value = await this._client.ReadAsync(4, this._disposeToken.Token);
								int readSize = value.ToInt32();
								byte[] payload = await this._client.ReadAsync(readSize, this._disposeToken.Token);
								this.CorrelatePayloadToRequest(payload);
							}
							catch (Exception ex)
							{
								if (!this._disposeToken.IsCancellationRequested)
								{
									this._log.ErrorFormat("Exception occured in polling read thread.  Exception={0}", new object[]
									{
										ex
									});
								}
							}
						}
					}
				}
				finally
				{
					Interlocked.Decrement(ref this._ensureOneActiveReader);
					this._log.DebugFormat("Closed down connection to: {0}", new object[]
					{
						this._client.Endpoint
					});
				}
			}, TaskCreationOptions.LongRunning);
		}
        private void CorrelatePayloadToRequest(byte[] payload)
        {
            int num = payload.Take(4).ToArray<byte>().ToInt32();
            KafkaConnection.AsyncRequestItem asyncRequestItem;
            if (this._requestIndex.TryRemove(num, out asyncRequestItem))
            {
                asyncRequestItem.ReceiveTask.SetResult(payload);
                return;
            }
            this._log.WarnFormat("Message response received with correlationId={0}, but did not exist in the request queue.", new object[]
			{
				num
			});
        }
        private int NextCorrelationId()
        {
            int num = Interlocked.Increment(ref this._correlationIdSeed);
            if (num > 2147482647)
            {
                Interlocked.Exchange(ref this._correlationIdSeed, 0);
            }
            return num;
        }
        private void ResponseTimeoutCheck()
        {
            try
            {
                this._timeoutSemaphore.Wait();
                List<KafkaConnection.AsyncRequestItem> list = (
                    from x in this._requestIndex.Values
                    where x.CreatedOnUtc.Add(this._responseTimeoutMS) < DateTime.UtcNow || this._disposeToken.IsCancellationRequested
                    select x).ToList<KafkaConnection.AsyncRequestItem>();
                foreach (KafkaConnection.AsyncRequestItem current in list)
                {
                    KafkaConnection.AsyncRequestItem asyncRequestItem;
                    if (this._requestIndex.TryRemove(current.CorrelationId, out asyncRequestItem))
                    {
                        if (this._disposeToken.IsCancellationRequested)
                        {
                            asyncRequestItem.ReceiveTask.TrySetException(new ObjectDisposedException("The object is being disposed and the connection is closing."));
                        }
                        else
                        {
                            asyncRequestItem.ReceiveTask.TrySetException(new ResponseTimeoutException(string.Format("Timeout Expired. Client failed to receive a response from server after waiting {0}ms.", this._responseTimeoutMS), new object[0]));
                        }
                    }
                }
            }
            finally
            {
                this._timeoutSemaphore.Release();
            }
        }
        public void Dispose()
        {
            if (Interlocked.Increment(ref this._disposeCount) != 1)
            {
                return;
            }
            this._responseTimeoutTimer.End();
            this._disposeToken.Cancel();
            if (this._connectionReadPollingTask != null)
            {
                this._connectionReadPollingTask.Wait(TimeSpan.FromSeconds(1.0));
            }
            using (this._disposeToken)
            {
                this.ResponseTimeoutCheck();
            }
            using (this._client)
            {
                using (this._responseTimeoutTimer)
                {
                }
            }
        }
    }
}
